package mr;

/*
 * YU LIU : 2012-11-30 @ Kochi
 */
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.StringTokenizer;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hama.HamaConfiguration;

import util.LongPair;
import util.Tool;

public class Scan {

	public static class FirstMapper extends
			Mapper<LongWritable, Text, IntWritable, LongPair> {

		private final static IntWritable one = new IntWritable(1);

		private long sum = 0;
		private long lastLineKey = 0;

		protected void setup(Context context) throws IOException,
				InterruptedException {
			// get
		}

		public void map(LongWritable key, Text line, Context context)
				throws IOException, InterruptedException {
			String aline = line.toString().trim();
			if (aline.isEmpty())
				return;

			// String[] eles = aline.split(" ");
			// // int sum = 0;
			// for (int i = 0; i < eles.length; i++) {
			// sum += Integer.parseInt(eles[i]);
			// }
			StringTokenizer itr = new StringTokenizer(aline);
			while (itr.hasMoreTokens()) {
				sum += Integer.parseInt(itr.nextToken().trim());
			}

			lastLineKey = key.get();
			// context.write(one, new IntPair((int) key.get(), sum));

		}

		protected void cleanup(Context context) throws IOException,
				InterruptedException {
			context.write(one, new LongPair(lastLineKey, sum));
		}
	}

	public static class ScanReducer extends
			Reducer<IntWritable, LongPair, Text, NullWritable> {
		String interSum = "";

		private static final Log LOG = LogFactory.getLog(ScanReducer.class);

		public void reduce(IntWritable key, Iterable<LongPair> values,
				Context context) throws IOException, InterruptedException {
			ArrayList<LongPair> sorted = new ArrayList<LongPair>();

			long start = System.currentTimeMillis();
			// LOG.info("++++++++++++++++++++++++++++++++++++++++++++++" +
			// start);
			LOG.info("constructing list");
			for (LongPair val : values) {
				LongPair tmp = new LongPair(val.first, val.second);
				sorted.add(tmp);
			}
			LOG.debug("finish constructing list. Time ="
					+ (System.currentTimeMillis() - start));
			LOG.debug("++++++++++++++++++++++++++++++++++++++++++++++");

			start = System.currentTimeMillis();
			LOG.debug("sort list");
			Tool.sortList(sorted); // sorting, for dis-parallel cases
			LOG.debug("sort list. Time ="
					+ (System.currentTimeMillis() - start));

			start = System.currentTimeMillis();
			interSum = list2String(sorted);
			LOG.debug("to String. Time ="
					+ (System.currentTimeMillis() - start));
			// sorted.
			context.write(new Text(list2String(sorted)), NullWritable.get());

			start = System.currentTimeMillis();
			// LOG.info("++++++++++++++++++++++++++++++++++++++++++++++" +
			// start);

		}

		protected void cleanup(Context context) throws IOException,
				InterruptedException {
			// context.getConfiguration().set("Scan_test", interSum);
		}
	}

	private static String list2String(ArrayList<LongPair> sorted) {
		StringBuffer rst = new StringBuffer(1024);
		for (LongPair pr : sorted) {
			rst.append(pr.first + "=" + pr.second + ";");
		}
		return rst.toString();
	}

	// private static void sortList(ArrayList<LongPair> sorted) {
	// // sort
	// Collections.sort(sorted, new LongPairComparator());
	// }

	// public static class IDReducer extends
	// Reducer<LongWritable, Text, LongWritable, Text> {
	//
	// public void reduce(LongWritable key, Iterable<Text> values,
	// Context context) throws IOException, InterruptedException {
	// ArrayList<IntPair> sorted = new ArrayList<IntPair>();
	//
	// for (Text val : values) {
	// context.write(key, val);
	// }
	//
	// }
	// }

	/**
	 * @param args
	 */
	public static void main(String[] args) throws Exception {

		Boolean ok = false;
		Path OUT_TMP = null;
		if (args.length >= 2) {
			Path INPUT = new Path(args[0]);
			Path OUTPUT = new Path(args[1]);

			Configuration conf = new Configuration();
			Job job = new Job(conf, "scan1");
			job.setJarByClass(Scan.class);
			job.setMapperClass(FirstMapper.class);
			job.setReducerClass(ScanReducer.class);
			job.setNumReduceTasks(1);

			job.setMapOutputKeyClass(IntWritable.class);
			job.setMapOutputValueClass(LongPair.class);
			job.setOutputKeyClass(IntWritable.class);
			job.setOutputValueClass(Text.class);
			job.setOutputFormatClass(TextOutputFormat.class);
			FileInputFormat.addInputPath(job, new Path(args[0]));
			OUT_TMP = new Path(args[1] + "_int");
			FileOutputFormat.setOutputPath(job, OUT_TMP);
			ok = job.waitForCompletion(true);
			System.out.println("+++++++++++++++++++++++++++++++++");
		}

		if (ok) {

			Configuration conf2 = new Configuration();
			FileSystem fs = FileSystem.get(conf2);
			Path partialSumList = new Path(OUT_TMP, "part-r-00000");

			// String scanSum = fs.open(partialSumList).readLine().trim(); //
			// not very good
			Path localCache = new Path("/tmp/scan_mr/partialSumList.txt");
			fs.copyToLocalFile(partialSumList, localCache);

			Path cacheFile = new Path("scan/cache/partialSumList.txt");
			if (fs.exists(cacheFile))
				fs.delete(cacheFile, true);
			fs.copyFromLocalFile(localCache, cacheFile);
			// scanSum = scanSum.split("\t")[1];

			DistributedCache.addCacheFile(cacheFile.toUri(), conf2);
			System.out.println(cacheFile.toUri());

			// conf2.set("partialSumListURI", localCache.toString());
			Job job2 = new Job(conf2, "scan2");
			// job2.addCacheFile(partialSumList.toUri());
			job2.setJarByClass(SecondMapper.class);
			job2.setMapperClass(SecondMapper.class);
			job2.setNumReduceTasks(0);
			job2.setMapOutputKeyClass(Text.class);
			job2.setMapOutputValueClass(NullWritable.class);
			// job2.setOutputKeyClass(NullWritable.class);
			// job2.setOutputValueClass(Text.class);
			job2.setInputFormatClass(TextInputFormat.class);
			job2.setOutputFormatClass(TextOutputFormat.class);
			FileInputFormat.addInputPath(job2, new Path(args[0]));
			FileOutputFormat.setOutputPath(job2, new Path(args[1]));
			int rst = job2.waitForCompletion(true) ? 0 : 1;
			fs.deleteOnExit(OUT_TMP);
			System.exit(rst);

		} else
			System.exit(1);

	}

	static void printOutput(Configuration conf, Path file) throws IOException {
		FileSystem fs = FileSystem.get(conf);
		FileStatus[] files = fs.listStatus(file);
		for (int i = 0; i < files.length; i++) {
			if (files[i].getLen() > 0) {
				FSDataInputStream in = fs.open(files[i].getPath());
				IOUtils.copyBytes(in, System.out, conf, false);
				in.close();
				break;
			}
		}
		fs.delete(file, true);
	}
}
